package com.itc.access.listener;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.itc.access.server.*;
import com.itc.access.server.impl.NtfmfTmiServerImpl;
import com.itc.system.api.listener.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import com.huawei.it.eip.ump.client.consumer.ConsumeStatus;
import com.huawei.it.eip.ump.client.consumer.Consumer;
import com.huawei.it.eip.ump.client.consumer.MessageModel;
import com.huawei.it.eip.ump.client.listener.MessageListener;
import com.huawei.it.eip.ump.common.exception.UmpException;
import com.huawei.it.eip.ump.common.message.Message;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.io.UnsupportedEncodingException;
import java.util.LinkedList;
import java.util.List;

import org.springframework.beans.factory.annotation.Value;

import javax.annotation.PreDestroy;


@Component
public class ConsumerStartupListener {

    @Autowired
    private DynFlightServer dynFlightServer;

    @Autowired
    private ErrorInfoServer errorInfoServer;
    @Autowired
    private AtcFlightBaseMsgServer atcFlightBaseMsgServer;
    @Autowired
    private AtcFlightEffsMsgServer atcFlightEffsMsgServer;
    @Autowired
    private FlightOutSortMsgServer flightOutSortMsgServer;
    @Autowired
    private FlowControlMsgServer flowControlMsgServer;
    @Autowired
    private RunWayCapacityMsgServer runWayCapacityMsgServer;
    @Autowired
    private NtfmfCtiServer ntfmfCtiServer;
    @Autowired
    private NtfmfTmiServerImpl ntfmfTmiServer;
    @Autowired
    private CastAdsbServer castAdsbServer;
    @Autowired
    private CaaCfplaServer caaCfplaServer;

    @Autowired
    private FlightGuaRanteeService flightGuaRanteeService;

    @Value("${Mqs.url}")
    private String url;
    @Value("${Mqs.app-id}")
    private String AppId;
    @Value("${Mqs.app-secret}")
    private String AppSecret;
    @Value("${Mqs.app-Topic}")
    private String AppTopic;
    @Value("${Mqs.tags.dynflight}")
    private String dynflight;
    @Value("${Mqs.tags.atcflightbasemsg}")
    private String atcflightbasemsg;
    @Value("${Mqs.tags.atcflighteffsmsg}")
    private String atcflighteffsmsg;
    @Value("${Mqs.tags.flightoutsortmsg}")
    private String flightoutsortmsg;
    @Value("${Mqs.tags.flowcontrolmsg}")
    private String flowcontrolmsg;
    @Value("${Mqs.tags.runwaycapacitymsg}")
    private String runwaycapacitymsg;
    @Value("${Mqs.tags.ntfmfcti}")
    private String ntfmfcti;
    @Value("${Mqs.tags.ntfmftmi}")
    private String ntfmftmi;
    @Value("${Mqs.tags.castadsb}")
    private String castadsb;
    @Value("${Mqs.tags.caacfpla}")
    private String caacfpla;
    @Value("${Mqs.tags.guarantee}")
    private String guarantee;
    private static final ObjectMapper objectMapper = new ObjectMapper();

    private Consumer consumer;

    /**
     * 航班动态数据 dynflight
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void dynFlightListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(dynflight); // 航班动态数据
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("dynflight");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {
                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                DynFlight dynFlight = new DynFlight();

                try {
                    dynFlight = objectMapper.readValue(body, DynFlight.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();

                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");

                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);
                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }
                dynFlightServer.save(dynFlight);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * 进程单航班基本信息 ATCFLIGHTBASEMSG
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void atcFlightBaseMsgListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(atcflightbasemsg); // 进程单航班基本信息
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("atcflightbasemsg");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                AtcFlightBaseMsg atcFlightBaseMsg = new AtcFlightBaseMsg();

                try {
                    atcFlightBaseMsg = objectMapper.readValue(body, AtcFlightBaseMsg.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }
                atcFlightBaseMsgServer.save(atcFlightBaseMsg);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * 航班塔台保障信息 ATCFLIGHTEFFSMSG
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void atcFlightEffSmsgListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(atcflighteffsmsg); // 航班塔台保障信息
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("atcflighteffsmsg");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                AtcFlightEffsMsg atcFlightEffsMsg = new AtcFlightEffsMsg();

                try {
                    atcFlightEffsMsg = objectMapper.readValue(body, AtcFlightEffsMsg.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }
                atcFlightEffsMsgServer.save(atcFlightEffsMsg);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * 航班离港排序信息 FLIGHTOUTSORTMSG
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void flightOutSortMsgListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(flightoutsortmsg); // 航班离港排序信息
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("FLIGHTOUTSORTMSG");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {
                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                FlightOutSortMsg flightOutSortMsg = new FlightOutSortMsg();

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                try {
                    flightOutSortMsg = objectMapper.readValue(body, FlightOutSortMsg.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }
                flightOutSortMsgServer.save(flightOutSortMsg);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * 流量控制信息(合深圳、广州流控) FLOWCONTROLMSG
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void flowControlMsgListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(flowcontrolmsg); // 流量控制信息(合深圳、广州流控)
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("flowcontrolmsg");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                FlowControlMsg flowControlMsg = new FlowControlMsg();

                try {
                    flowControlMsg = objectMapper.readValue(body, FlowControlMsg.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }

                flowControlMsgServer.save(flowControlMsg);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * 流跑道运行模式信息 RUNWAYCAPACITYMSG
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void runWayCapacityMsgListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(runwaycapacitymsg); // 跑道运行模式信息
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("runwaycapacitymsg");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                RunWayCapacityMsg runWayCapacityMsg = new RunWayCapacityMsg();

                try {
                    runWayCapacityMsg = objectMapper.readValue(body, RunWayCapacityMsg.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }

                runWayCapacityMsgServer.save(runWayCapacityMsg);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * 航班协同时隙信息 NTFMFCTI
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void ntfmfCtiListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(ntfmfcti); // 航班协同时隙信息
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("ntfmfcti");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                NtfmfCti ntfmfCti = new NtfmfCti();

                try {
                    ntfmfCti = objectMapper.readValue(body, NtfmfCti.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }

                ntfmfCtiServer.save(ntfmfCti);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * 流量控制措施 NTFMFTMI
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void ntfmfTmiListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(ntfmftmi); // 流量控制措施
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("ntfmftmi");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                NtfmfTmi ntfmfTmi = new NtfmfTmi();

                try {
                    ntfmfTmi = objectMapper.readValue(body, NtfmfTmi.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }

                ntfmfTmiServer.save(ntfmfTmi);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * ADSB数据 CASTADSB
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void CastAdsbListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(castadsb); // ADSB数据
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("castadsb");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                ObjectMapper objectMapper = new ObjectMapper();

                try {
                    List<CastAdsb> castAdsbList = objectMapper.readValue(body, new TypeReference<List<CastAdsb>>() {});
                    List<CastAdsb> saveList = new LinkedList<>();
                    for (CastAdsb castAdsb : castAdsbList) {
                        // 处理每个 CastAdsb 对象
                        saveList.add(castAdsb);
                    }
                    if(saveList.size() > 0){
                        castAdsbServer.saveBatch(saveList);
                    }
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }
                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * 机场保障计划信息(民航) CAACFPLA
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void CaaCfplaListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(caacfpla); // 机场保障计划信息(民航)
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("caacfpla");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                CaaCfpla caaCfpla = new CaaCfpla();

                try {
                    caaCfpla = objectMapper.readValue(body, CaaCfpla.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }

                caaCfplaServer.save(caaCfpla);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    /**
     * 航班保障节点信息接入
     * @throws UmpException
     */
    @EventListener(ApplicationReadyEvent.class)
    public void guaranteeListener() throws UmpException {
        // 发送消息的topic
        Consumer consumer = new Consumer();
        consumer.setUmpNamesrvUrls(url); // 设置MQS的服务器地址,多个IP使用;进行分隔
        consumer.setAppId(AppId); // 设置客户端账号(appId/应用ID)
        consumer.setAppSecret(AppSecret); // 设置客户端密钥(appSecret/应用密钥)
        consumer.setTopic(AppTopic); // 设置Topic名称(Topic)
        consumer.setEncryptTransport(true); // 设置是否需要加密传输
        consumer.setTags(guarantee); // 机场保障计划信息(民航)
        consumer.setGroupWithTags(true); // 如果同一个appid在一个集群上订阅不同的tag，需要设置 groupWithTags 为 true
        consumer.setMessageModel(MessageModel.CLUSTERING); // 集群模式(默认模式)
        ErrorInfo errorInfo = new ErrorInfo();
        errorInfo.setTags("guarantee");
        consumer.subscribe(new MessageListener() {
            public ConsumeStatus consume(Message message) throws UnsupportedEncodingException {

                message.getMessageId(); //messageId为每条数据发布成功后系统生成的唯一标识，可用来做数据是否发送成功核对使用

                // 消费消息的业务逻辑
                byte[] body = message.getBody();
                FlightGuaRantee guarantee = new FlightGuaRantee();

                try {
                    guarantee = objectMapper.readValue(body, FlightGuaRantee.class);
                } catch (IOException e) {
                    StringWriter sw = new StringWriter();
                    PrintWriter pw = new PrintWriter(sw);
                    e.printStackTrace(pw);
                    String exceptionAsString = sw.toString();
                    String result = message.getTags() + "Receive: " + new String(message.getBody(), "UTF-8");
                    errorInfo.setErrorInfo(exceptionAsString);
                    errorInfo.setBodyInfo(result);
                    errorInfoServer.save(errorInfo);

                    return ConsumeStatus.RECONSUME_LATER; // 反序列化失败，返回重试
                }

                flightGuaRanteeService.save(guarantee);

                // 正常接收到消息后，请务必返回CONSUME_SUCCESS，只有在业务处理失败才返回RECONSUME_LATER
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start(); // 启动消费者，建议在应用程序关闭的时候关闭。
    }

    @PreDestroy
    public void onShutdown() throws UmpException {
        if (consumer != null) {
            consumer.shutdown(); // 关闭消费者
        }
    }
}
